#include <common.h>
#include <mqtt_paho.h>
#include <unistd.h>
#include <string.h>
#include <system.h>
#include <pthread.h>
#include <signal.h>
#include <MQTTClient.h>

static MQTTClient client;
static int last_token;
static int retained;
static int mqttpaho_subscribe_ok;
static int mqttpaho_enable;
static int mqtt_reconnect_flag;

void mqtt_paho_enable(void)
{
	mqttpaho_enable = 1;
}

void mqtt_paho_disable(void)
{
	mqttpaho_enable = 0;
}

static void* mqtt_client_handle_thread(void *arg)
{
	struct mqtt_client *mqtt_cli = arg;
	char* topicName = NULL;
	char url[256] = {0};
	char subTop[256] = {0};
	char username[128] = {0};
	char password[128] = {0};
	char clientid[128] = {0};

	int rc = 0;
	int topicLen = 0;
	MQTTClient_message* message = NULL;
	
	MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
	sprintf(url, "%s:%d", mqtt_cli->host, mqtt_cli->port);

	MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
	conn_opts.keepAliveInterval = mqtt_cli->keepAlive;
	conn_opts.username = username;
	conn_opts.password = password;

	while (1)
	{
		rc = MQTTClient_receive(client, &topicName, &topicLen, &message, 5000);
		if (message)
		{
			if(mqtt_cli->mqtt_msg_handle_callback)
			{
				mqtt_cli->mqtt_msg_handle_callback(topicName, message->payload, message->payloadlen);
			}
			
			MQTTClient_freeMessage(&message);
			MQTTClient_free(topicName);
			message = NULL;
		}

		if (rc != 0 || mqtt_reconnect_flag)
		{
			mqtt_reconnect_flag = 0;
			mqttpaho_subscribe_ok = 0;

			/* reconnect mqtt server */
			mqtt_client_info_init(username, password, clientid, mqtt_cli->gwid);
			DEBUG("MQTTClient_disconnect");
			MQTTClient_disconnect(client, 0);
			DEBUG("MQTTClient_destroy");
			MQTTClient_destroy(&client);
			client = NULL;
			MQTTClient_createWithOptions(&client, url, clientid, MQTTCLIENT_PERSISTENCE_NONE, NULL, &createOpts);
			DEBUG("MQTTClient_connect");
			rc = MQTTClient_connect(client, &conn_opts);
			if(rc == MQTTREASONCODE_SUCCESS) 
			{
				//平台请求设置Topic
				sprintf(subTop, "/%s/request/control", mqtt_cli->gwid);
				DEBUG("subTop=%s", subTop);
				if(MQTTClient_subscribe(client, subTop, mqtt_cli->qos))
				{
					DEBUG("MQTTClient_subscribe error topic=%s", subTop);
				}

				sprintf(subTop, "/all/request/control");
				DEBUG("subTop=%s", subTop);
				if(MQTTClient_subscribe(client, subTop, mqtt_cli->qos))
				{
					DEBUG("MQTTClient_subscribe error topic=%s", subTop);
				}

				//网关请求通知回应Topic
				//sprintf(subTop, "/%s/respond/notify", mqtt_cli->gwid);
				//DEBUG("subTop=%s", subTop);
				//if(MQTTClient_subscribe(client, subTop, mqtt_cli->qos))
				//{
				//	DEBUG("MQTTClient_subscribe error topic=%s", subTop);
				//}

				DEBUG("mqtt connected success %s", url);
				mqttpaho_subscribe_ok = 1;
				mqtt_cli->mqtt_connect_handle_callback();
			} 			
			else
			{
				DEBUG("mqtt connected error rc=%d,url=%s", rc, url);
				second_sleep(60);
			}
		}
	}

	MQTTClient_disconnect(client, 0);
	MQTTClient_destroy(&client);
	return NULL;
}

int mqtt_paho_connect(void)
{
	return mqttpaho_subscribe_ok;
}

int mqtt_paho_publish(char *topic, unsigned char *msg, int msglen, int qos)
{
	if(mqtt_paho_connect())
	{
		return MQTTClient_publish(client, topic, msglen, msg, qos, retained, &last_token);
	}

	return -1;
}

int mqtt_paho_subscribe(char *topic, int subscribe, int qos)
{
	if(client == NULL || mqttpaho_enable == 0)
	{
		return -1;
	}

	if(subscribe)
	{
		//主题订阅
		MQTTClient_subscribe(client, topic, qos);
	}
	else
	{
		//取消订阅
		MQTTClient_unsubscribe(client, topic);
	}
	
	return 0;
}

int mqtt_paho_start(struct mqtt_client * opts)
{
	create_thread(mqtt_client_handle_thread, opts);	
	return 0;
}

void mqtt_paho_reconnect(void)
{
	mqtt_reconnect_flag = 1;
}

